package com.ruoyi.system.webchat;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.ruoyi.common.config.AppConfig;
import com.ruoyi.common.constant.CacheConstants;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.enums.message.MessageIsRead;
import com.ruoyi.common.enums.session.ConversationType;
import com.ruoyi.common.utils.SecurityUtils;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.common.utils.spring.SpringUtils;
import com.ruoyi.common.utils.taskscheduler.TaskScheduler;
import com.ruoyi.common.utils.uuid.IdUtils;
import com.ruoyi.system.domain.vo.MessageVO;
import com.ruoyi.system.domain.vo.SessionVO;
import com.ruoyi.system.service.ISessionService;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@ServerEndpoint(value = "/chat/{userId}")
@Component
@Slf4j
public class WebChatServer {

    // 存储所有在线连接，key:用户名，value:WebSocket会话
    private static final ConcurrentHashMap<Long, Session> onlineSessions = new ConcurrentHashMap<>();
    // 存储聊天历史  key:sessionId，value:消息列表
    private static ConcurrentHashMap<String, List<MessageVO>> chatHistoryMap = new ConcurrentHashMap<>();


    private RedisCache redisCache = SpringUtils.getBean(RedisCache.class);

    private ISessionService sessionService = SpringUtils.getBean(ISessionService.class);


    private TaskScheduler taskScheduler = SpringUtils.getBean(TaskScheduler.class);


    /**
     * TODO 消息保存线程池(CPU IO密集型任务) - 用于异步保存消息
     *      定时任务线程池
     */

    /**
     * 消息发送线程池(CPU 密集型任务) - 用于异步发送消息
     */
    private static final ExecutorService messageSenderExecutor = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors(), // 获取系统CPU核数
            new ThreadFactory() {
                private final AtomicInteger conter = new AtomicInteger(1);
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "messageSender-thread-" + conter.getAndIncrement());
                }
            }
    );

    /**
     * WebSocket 会话的业务信息（仅包含需要序列化的字段）
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class WebSocketSessionInfo {
        private String sessionId;      // 会话ID
        private Long userId;           // 用户ID
        private LocalDateTime connectTime; // 连接时间
    }


    // 连接建立时调用
    @OnOpen
    public void onOpen(Session session,@PathParam("userId") Long userId) {
        // 将新用户加入在线列表
        log.info("后端第一步：用户{}连接成功", userId);
        onlineSessions.put(userId, session);
        Map<String, WebSocketSessionInfo> cacheMap = redisCache.getCacheMap(CacheConstants.USER_ONLINE);


        if(!ObjectUtils.isEmpty(cacheMap)){
            if(cacheMap.containsKey(String.valueOf(userId))){
                log.info("用户已存在");
            }else {
                WebSocketSessionInfo info = new WebSocketSessionInfo();
                info.setSessionId(session.getId());
                info.setUserId(userId);
                info.setConnectTime(LocalDateTime.now());

                cacheMap.put(String.valueOf(userId),info);
                redisCache.setCacheMap(CacheConstants.USER_ONLINE,cacheMap,AppConfig.CACHE_TTL, TimeUnit.SECONDS);
            }
        }else {
            cacheMap = new HashMap<>();
            WebSocketSessionInfo info = new WebSocketSessionInfo();
            info.setSessionId(session.getId());
            info.setUserId(userId);
            info.setConnectTime(LocalDateTime.now());

            cacheMap.put(String.valueOf(userId),info);
            redisCache.setCacheMap(CacheConstants.USER_ONLINE,cacheMap,AppConfig.CACHE_TTL, TimeUnit.SECONDS);
        }
    }

    // 收到客户端消息时调用
    @OnMessage
    public void onMessage(String message, Session session) throws UnsupportedEncodingException {
        log.info("后端第二步：用户连接成功:{}", message);
//        message = "{\"conversationId\":\"2b32d564c8924dc8bfe54b6a23aabe9a\",\"senderId\":1,\"receiverId\":101,\"messageType\":1,\"content\":\"2222\"}";
        MessageVO messageVO = JSON.parseObject(message, new TypeReference<MessageVO>() {
        });

        SessionVO queryVO = new SessionVO();
        queryVO.setId(messageVO.getConversationId());
        List<com.ruoyi.system.domain.Session> sessions = sessionService.selectSessionList(queryVO);
        SessionVO sessionVO = new SessionVO();
        if(CollectionUtils.isEmpty(sessions)){
            log.info("sessionId 不正确");
            return;
        }

//        messageVO.setSenderStatus(MsgSendStatus.DOING.getCode());
//        messageVO.setIsRead(MessageIsRead.NO.getCode());
//        messageVO.setSendTime(new Date());

        // 设置基本信息
        messageVO.setIsSync(0);
        messageVO.setCreator(String.valueOf(messageVO.getSenderId()));  // TODO 获取用户信息待完善
        messageVO.setIsDeleted(0);
        messageVO.setVersion(0L);
        messageVO.setCreateTime(new Date());
        messageVO.setUpdateTime(new Date());


        BeanUtils.copyProperties(sessions.get(0),sessionVO);
        messageVO.setSessionVO(sessionVO);

        messageVO.setId(IdUtils.simpleUUID());

        Long receiverId = messageVO.getReceiverId();
        Integer conversationType = messageVO.getSessionVO().getConversationType();


        if(ConversationType.SINGLE.getCode().equals(conversationType)){

//            Session receiverSession = redisCache.getCacheMapValue(CacheConstants.USER_ONLINE, String.valueOf(receiverId));
            Session receiverSession = onlineSessions.get(receiverId);
            if(!ObjectUtils.isEmpty(receiverSession)){

                // 是否已读判断  当用户在线，同时点击了当前会话产生的会话，才能被设置为已读，否则都是未读的情况
                boolean isRead = checkReadMsg(receiverId,messageVO.getSessionVO().getId());
                log.info("onMessage 消息是否已读:{}",isRead);
                handleMessageInfo(messageVO,isRead,false);
            }else {
                log.info("用户未上线  正在做离线处理");
                handleMessageInfo(messageVO,false,false);
            }
        }
        else if(ConversationType.GROUP.getCode().equals(conversationType)){
            // 群聊

        }

    }

    private boolean checkReadMsg(Long receiverId, String sessionId) {
        return redisCache.getCacheMapValue(CacheConstants.USER_CUR_SESSION + receiverId, sessionId);
    }

    private void handleMessageInfo(MessageVO messageVO, Boolean isRead, boolean isGroup) {
        messageVO.setIsRead(isRead?MessageIsRead.YES.getCode():MessageIsRead.NO.getCode());
        if (messageVO.getMessageType() != null) {

            switch (messageVO.getMessageType()){
                case 1:
                    // 文字
                    if(isGroup){
                        sendToAllClients(messageVO);
                    }else {
                        //设置消息信息到redis中 TODO  是否需要把消息转化成字符串？
//                        String msgVersion = redisCache.getCacheObject(CacheConstants.MSG_VERSION);
//                        String curValidMsgKey = redisCache.getCurValidMsgKey(msgVersion);

                        // TODO 睡觉  看效果
//                        try {
//                            Thread.sleep(1000 * 30);
//                        } catch (InterruptedException e) {
//                            throw new RuntimeException(e);
//                        }

                        redisCache.setCacheObject(CacheConstants.MSG_KEY + messageVO.getId(),messageVO, AppConfig.CACHE_TTL, TimeUnit.MINUTES);  // 过期时间  60分钟

                        // 设置当前用户会话列表  user:sessionId:curMsgs:{接收者ID}:{sessionId}
                        String userSessionCurMsgListKey = StringUtils.handleString(CacheConstants.USER_SESSION_CUR_MSG_LIST, messageVO.getConversationId(), String.valueOf(messageVO.getReceiverId()));
                        log.info("handleMessageInfo userSessionCurMsgListKey:{}",userSessionCurMsgListKey);
                        redisCache.setZCacheObject(userSessionCurMsgListKey,messageVO.getSendTime().getTime(),messageVO.getId());

                        //设置用户未读消息 user:sessionId:unreadList:{接收者ID}:{sessionId}  这里的逻辑在 ChatPage.vue的 updateSessionVOInfo 方法调用

//                        if(!isRead){
//                            //设置用户未读消息 user:sessionId:unreadList:{接收者ID}:{sessionId}
//                            String userUnreadKey = StringUtils.handleString(CacheConstants.USER_UNREAD_LIST, messageVO.getConversationId(), String.valueOf(messageVO.getReceiverId()));
//                            log.info("handleMessageInfo userUnreadKey:{}",userUnreadKey);
//                            setDataToCache(messageVO, userUnreadKey);
//                        }
                        sendPrivateMessage(messageVO);
                    }
                    break;
                case 2:
                    // 图片
                    break;
                case 3:
                    // 文件
                    break;
                case 4:
                    // 引用消息

                    if (StringUtils.isNotEmpty(messageVO.getQuoteMessageId())) {


                    }

                    break;
                case 5:
                    // 转发消息
                    if(StringUtils.isNotEmpty(messageVO.getForwardFromId())){

                    }
                    break;
                default:
                    log.info("未知的消息类型");
            }
        }
    }

    private void sendToAllClients(MessageVO messageVO) {
        Long senderKey = messageVO.getSenderId();
//        Map<String, Session> cacheMap = redisCache.getCacheMap(CacheConstants.USER_ONLINE);
//        Set<Map.Entry<String, Session>> entries = cacheMap.entrySet();
        Set<Map.Entry<Long, Session>> entries = onlineSessions.entrySet();

        for (Map.Entry<Long, Session> entry : entries) {
            // TODO 如果message是广播消息，不发送给自己  同时  还要发送同会话的群成员  同时还要分 已读、未读情况

//            long receiverId = Long.parseLong(entry.getKey());
            long receiverId = entry.getKey();
            if(ConversationType.GROUP.getCode().equals(messageVO.getSessionVO().getConversationType())
                    && receiverId == senderKey){
                // 是群聊，并且遍历到了自己  不发送消息
                continue;
            }
            // 发送消息
            Session receiver = entry.getValue();
            if (receiver != null && receiver.isOpen()) {
                MessageVO messageTemp = new MessageVO();
                BeanUtils.copyProperties(messageVO,messageTemp);
                messageTemp.setReceiverId(receiverId);
                String messageStr = JSON.toJSONString(messageTemp);
                messageSenderExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        sendToSingleClient(receiver,messageStr );
                    }
                });
            }
        }
    }

    private void sendPrivateMessage(MessageVO messageVO) {
        Long receiverId = messageVO.getReceiverId();
//        Session receiverSession = redisCache.getCacheMapValue(CacheConstants.USER_ONLINE, String.valueOf(receiverId));
        Session receiverSession = onlineSessions.get(receiverId);
        // 再次检验  客户端是否上线
        if (receiverSession != null && receiverSession.isOpen()) {
            String messageStr = JSON.toJSONString(messageVO);
            messageSenderExecutor.submit(()->{
                sendToSingleClient(receiverSession, messageStr);
            });
        }else {
            log.info("用户未上线  正在做离线处理");
        }
    }

    private static void sendToSingleClient(Session receiverSession, String messageStr) {
        if (receiverSession != null && receiverSession.isOpen()) {
            try {
                log.info("后端第三步：发送消息给用户：{}",receiverSession.getId());
                receiverSession.getBasicRemote().sendText(messageStr);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }else {
            log.info("用户未上线  正在做离线处理");
        }
    }

    // 发生错误时调用
    @OnError
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }

    // 连接关闭时调用
    @OnClose
    public void onClose(@PathParam("userId") Long userId, Session session) {
        // 从在线列表移除
//        Session session1 = redisCache.getCacheMapValue(CacheConstants.USER_ONLINE, String.valueOf(userId));
        Session session1 = onlineSessions.get(userId);
        if (session1 != null) {
            try {
                session1.close();
                session.close();

                onlineSessions.remove(userId);
                redisCache.deleteCacheMapValue(CacheConstants.USER_ONLINE, String.valueOf(userId));
                redisCache.deleteObject(CacheConstants.USER_CUR_SESSION + SecurityUtils.getUserId());

                log.info("WebChatServer onClose 用户：{} 退出",userId);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }


}
